//package com.huatai.datacenter.service.impl;
//
//import com.alibaba.nacos.client.naming.utils.CollectionUtils;
//import com.google.common.base.Charsets;
//import com.google.common.base.Predicate;
//import com.google.common.base.Throwables;
//import com.google.common.collect.Iterables;
//import com.google.common.collect.Lists;
//import com.google.common.collect.Maps;
//import com.google.common.collect.Sets;
//import com.huatai.datacenter.constant.Constants;
//import com.huatai.datacenter.entity.messagequeuemonitor.*;
//import com.huatai.datacenter.service.RMQMonitorService;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
//import org.apache.rocketmq.client.consumer.PullResult;
//import org.apache.rocketmq.client.exception.MQBrokerException;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.MQVersion;
//import org.apache.rocketmq.common.MixAll;
//import org.apache.rocketmq.common.admin.ConsumeStats;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.message.MessageQueue;
//import org.apache.rocketmq.common.protocol.body.*;
//import org.apache.rocketmq.common.protocol.route.BrokerData;
//import org.apache.rocketmq.remoting.exception.RemotingException;
//import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
//import org.apache.rocketmq.tools.admin.MQAdminExt;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.PostConstruct;
//import java.util.*;
//import java.util.concurrent.*;
//import java.util.concurrent.atomic.AtomicLong;
//import java.util.stream.Collectors;
//
///**
// * 最新版消息队列监控服务（RMQ） TODO 需要获取真实环境的RMQ NamesrvAddr地址，目前先用本地的
// * <a href="https://github.com/apache/rocketmq-dashboard/tree/master">...</a>
// *
// * @author lizhong
// */
//@Slf4j
//@Service
//public class RMQMonitorServiceImpl implements RMQMonitorService {
//
//	// RocketMQ的
//	private MQAdminExt mqAdminExt;
//
//	// 辅助线程池
//	private ExecutorService executorService;
//
//	// RMQ key: topic+group+brokerAddr  value: 该broker的所有分区offset情况
//	public Map<String/*brokerAddr*/, List<QueueOffset>> mapToQueueOffset1 = new HashMap<>();
//
//	public Map<String/*brokerAddr*/, String> RMQAddrToName = new HashMap<>();
//
//	// TODO 需要获取真实环境的RMQ NamesrvAddr地址，目前先用本地的!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
//	String namesrvAddr = "localhost:9876";
//
//	/**
//	 * 初始化executorService
//	 */
//	private void initExecutorService() {
//		// 初始化executorService
//		Runtime runtime = Runtime.getRuntime();
//		int corePoolSize = Math.max(10, runtime.availableProcessors() * 2);
//		int maximumPoolSize = Math.max(20, runtime.availableProcessors() * 2);
//		ThreadFactory threadFactory = new ThreadFactory() {
//			private final AtomicLong threadIndex = new AtomicLong(0);
//
//			@Override
//			public Thread newThread(Runnable r) {
//				return new Thread(r, "QueryGroup_" + this.threadIndex.incrementAndGet());
//			}
//		};
//		RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardOldestPolicy();
//		// 初始化executorService
//
//		ExecutorService executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 60L, TimeUnit.SECONDS,
//			new LinkedBlockingQueue<>(5000), threadFactory, handler);
//
//		this.executorService = executorService;
//	}
//
//	/**
//	 * 初始化MQAdminExt，TODO 需要NamesrvAddr
//	 */
//	private void initMQAdminExt() {
//
//		DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
//		mqAdminExt.setNamesrvAddr(namesrvAddr);
//		try {
//			mqAdminExt.start();
//		} catch (MQClientException e) {
//			log.error("create RMQ MQAdmin instance {} error!", mqAdminExt);
//			throw new RuntimeException(e);
//		}
//		log.info("create RMQ MQAdmin instance {} success.", mqAdminExt);
//		this.mqAdminExt = mqAdminExt;
//	}
//
//	/**
//	 * Bean后初始化!!
//	 */
//	@PostConstruct
//	public void init() {
//		initExecutorService();
//		initMQAdminExt();
//	}
//
//	// 系统的消费组集合
//	private static final Set<String> SYSTEM_GROUP_SET = new HashSet<>();
//
//	static {
//		SYSTEM_GROUP_SET.add(MixAll.TOOLS_CONSUMER_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.FILTERSRV_CONSUMER_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.SELF_TEST_CONSUMER_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.ONS_HTTP_PROXY_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_PULL_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_PERMISSION_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_OWNER_GROUP);
//		SYSTEM_GROUP_SET.add(MixAll.CID_SYS_RMQ_TRANS);
//	}
//
//	/**
//	 * 列出所有消费组列表 RMQ
//	 *
//	 * @param skipSysGroup 跳过系统组
//	 * @return 消费组信息列表
//	 */
//	public List<GroupConsumeInfo> getGroupListRMQ(boolean skipSysGroup) {
//		Set<String> consumerGroupSet = Sets.newHashSet();
//		try {
//			ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
//			System.out.println("examineBrokerClusterInfo: " + clusterInfo.toString());
//			for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
//				SubscriptionGroupWrapper subscriptionGroupWrapper = mqAdminExt.getAllSubscriptionGroup(brokerData.selectBrokerAddr(), 3000L);
//				consumerGroupSet.addAll(subscriptionGroupWrapper.getSubscriptionGroupTable().keySet());
//			}
//		} catch (Exception err) {
//			log.error("examineBrokerClusterInfo exception: {}", mqAdminExt, err);
//			Throwables.throwIfUnchecked(err);
//			throw new RuntimeException(err);
//		}
//		List<GroupConsumeInfo> groupConsumeInfoList = Collections.synchronizedList(Lists.newArrayList());
//		CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size());
//		for (String consumerGroup : consumerGroupSet) {
//			executorService.submit(() -> {
//				try {
//					GroupConsumeInfo consumeInfo = queryGroupRMQ(consumerGroup);
//					groupConsumeInfoList.add(consumeInfo);
//				} catch (Exception e) {
//					log.error("queryGroup exception, consumerGroup: {}", consumerGroup, e);
//				} finally {
//					countDownLatch.countDown();
//				}
//			});
//		}
//		try {
//			countDownLatch.await(30, TimeUnit.SECONDS);
//		} catch (InterruptedException e) {
//			log.error("query consumerGroup countDownLatch await Exception", e);
//		}
//
//		if (!skipSysGroup) {
//			groupConsumeInfoList.stream().map(group -> {
//				if (SYSTEM_GROUP_SET.contains(group.getGroup())) {
//					group.setGroup(String.format("%s%s", "%SYS%", group.getGroup()));
//				}
//				return group;
//			}).collect(Collectors.toList());
//		}
//		Collections.sort(groupConsumeInfoList);
//		return groupConsumeInfoList;
//	}
//
//	/**
//	 * 消费组名->消费组信息 RMQ
//	 *
//	 * @param consumerGroup 消费组名
//	 * @return
//	 */
//	public GroupConsumeInfo queryGroupRMQ(String consumerGroup) {
//		GroupConsumeInfo groupConsumeInfo = new GroupConsumeInfo();
//		try {
//			ConsumeStats consumeStats = null;
//			try {
//				consumeStats = mqAdminExt.examineConsumeStats(consumerGroup);
//			} catch (Exception e) {
//				log.warn("examineConsumeStats exception to consumerGroup {}, response [{}]", consumerGroup, e.getMessage());
//			}
//
//			ConsumerConnection consumerConnection = null;
//			try {
//				consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
//			} catch (Exception e) {
//				log.warn("examineConsumeStats exception to consumerGroup {}, response [{}]", consumerGroup, e.getMessage());
//			}
//
//			groupConsumeInfo.setGroup(consumerGroup);
//
//			if (consumeStats != null) {
//				groupConsumeInfo.setConsumeTps((int) consumeStats.getConsumeTps());
//				groupConsumeInfo.setDiffTotal(consumeStats.computeTotalDiff());
//			}
//
//			if (consumerConnection != null) {
//				groupConsumeInfo.setCount(consumerConnection.getConnectionSet().size());
//				groupConsumeInfo.setVersion(MQVersion.getVersionDesc(consumerConnection.computeMinVersion()));
//			}
//		} catch (Exception e) {
//			log.warn("examineConsumeStats or examineConsumerConnectionInfo exception, " + consumerGroup, e);
//		}
//		return groupConsumeInfo;
//	}
//
//	/**
//	 * - (主题, 消费组名)->消费状态列表 RMQ
//	 *
//	 * @param topic     主题
//	 * @param groupName 消费组名
//	 * @return 消费状态列表
//	 */
//	private List<TopicConsumerInfo> getConsumeStatsListRMQ(final String topic, String groupName) {
//		ConsumeStats consumeStats = null;
//		try {
//			// 限定消费组和主题
//			consumeStats = mqAdminExt.examineConsumeStats(groupName, topic);
//		} catch (Exception e) {
//			Throwables.throwIfUnchecked(e);
//			throw new RuntimeException(e);
//		}
//		List<MessageQueue> mqList = Lists.newArrayList(Iterables.filter(consumeStats.getOffsetTable().keySet(), new Predicate<MessageQueue>() {
//			@Override
//			public boolean apply(MessageQueue o) {
//				return StringUtils.isBlank(topic) || o.getTopic().equals(topic);
//			}
//		}));
//		Collections.sort(mqList);
//		List<TopicConsumerInfo> topicConsumerInfoList = Lists.newArrayList();
//		TopicConsumerInfo nowTopicConsumerInfo = null;
//		Map<MessageQueue, String> messageQueueClientMap = getClientConnection(groupName);
//
//		// 一个消费组一个主题下的所有队列，接下来就是要按brokername分组
//		for (MessageQueue mq : mqList) {
//			if (nowTopicConsumerInfo == null || (!StringUtils.equals(mq.getTopic(), nowTopicConsumerInfo.getTopic()))) {
//				nowTopicConsumerInfo = new TopicConsumerInfo(mq.getTopic(), groupName);
//				topicConsumerInfoList.add(nowTopicConsumerInfo);
//			}
//			QueueStatInfo queueStatInfo = QueueStatInfo.fromOffsetTableEntry(mq, consumeStats.getOffsetTable().get(mq));
//			queueStatInfo.setClientInfo(messageQueueClientMap.get(mq));
//			queueStatInfo.setLag(queueStatInfo.getBrokerOffset() - queueStatInfo.getConsumerOffset());
//			queueStatInfo.setBrokerName(mq.getBrokerName());
//
//			nowTopicConsumerInfo.appendQueueStatInfoRMQ(queueStatInfo);
//		}
//		return topicConsumerInfoList;
//	}
//
//	/**
//	 * - 消费组名->客户端连接?
//	 *
//	 * @param groupName 消费组名
//	 * @return 客户端连接
//	 */
//	private Map<MessageQueue, String> getClientConnection(String groupName) {
//		Map<MessageQueue, String> results = Maps.newHashMap();
//		try {
//			ConsumerConnection consumerConnection = mqAdminExt.examineConsumerConnectionInfo(groupName);
//			for (Connection connection : consumerConnection.getConnectionSet()) {
//				String clinetId = connection.getClientId();
//				ConsumerRunningInfo consumerRunningInfo = mqAdminExt.getConsumerRunningInfo(groupName, clinetId, false);
//				for (MessageQueue messageQueue : consumerRunningInfo.getMqTable().keySet()) {
////                    results.put(messageQueue, clinetId + " " + connection.getClientAddr());
//					results.put(messageQueue, clinetId);
//				}
//			}
//		} catch (Exception err) {
//			log.error("op=getClientConnection_error", err);
//		}
//		return results;
//	}
//
//	/**
//	 * 列出所有主题列表(不包括系统的主题) RMQ
//	 *
//	 * @param skipSysProcess  默认false
//	 * @param skipRetryAndDlq 默认false
//	 * @return 主题列表类
//	 */
//	public Set<String> getTopicListRMQ(boolean skipSysProcess, boolean skipRetryAndDlq) {
//		try {
//			// 通过MQAdminExt获取所有主题列表
//			TopicList allTopics = mqAdminExt.fetchAllTopicList();
//			// 先不考虑暴露出来系统主题，需要的去https://github.com/apache/rocketmq-dashboard/blob/master/src/main/java/org/apache/rocketmq/dashboard/service/impl/TopicServiceImpl.java#L63
//
//			// 一些加工处理
//			Set<String> topics =
//				allTopics.getTopicList().stream().map(topic -> {
//					if (!skipSysProcess/* && sysTopics.getTopicList().contains(topic)*/) {
//						topic = String.format("%s%s", "%SYS%", topic);
//					}
//					return topic;
//				}).filter(topic -> {
//					if (skipRetryAndDlq) {
//						return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
//							|| topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
//					}
//					return true;
//				}).collect(Collectors.toSet());
//			allTopics.getTopicList().clear();
//			allTopics.getTopicList().addAll(topics);
//			return allTopics.getTopicList();
//		} catch (Exception e) {
//			Throwables.throwIfUnchecked(e);
//			throw new RuntimeException(e);
//		}
//	}
//
//	/**
//	 * 主题->所有消费者 RMQ
//	 *
//	 * @param topic 主题
//	 * @return
//	 */
//	public Set<String> getGroupListByTopicRMQ(String topic) {
//		try {
//			GroupList groupList = mqAdminExt.queryTopicConsumeByWho(topic); // TODO BUG点
//			return groupList.getGroupList();
//		} catch (InterruptedException | MQBrokerException | RemotingException | MQClientException e) {
//			throw new RuntimeException(e);
//		}
//	}
//
//	/**
//	 * 主题->(消费组名, 消费信息(总堆积和各个队列的情况)) RMQ
//	 *
//	 * @param topic 主题
//	 * @return 该主题下所有消费组的消费情况（显示最小粒度为broker的队列）
//	 */
//	public List<LagStatVO> getConsumeStatsListByTopicRMQ(String topic) {
//
//		// 该主题的所有消费组所有broker的堆积情况
//		List<LagStatVO> lagStatVOList = new ArrayList<>();
//
//		try {
//			// 集群信息
//			ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
//
//			// 根据主题获取所有的消费者
//			Set<String> groupList = getGroupListByTopicRMQ(topic);
//
//			// 第二层遍历：遍历该主题下的所有消费者
//			for (String group : groupList) {
//				List<TopicConsumerInfo> topicConsumerInfoList = null;
//				try {
//					// 限定了topic和group
//					topicConsumerInfoList = getConsumeStatsListRMQ(topic, group);
//
//				} catch (Exception ignore) {
//				}
//
//				TopicConsumerInfo temp = CollectionUtils.isEmpty(topicConsumerInfoList) ? new TopicConsumerInfo(topic, group) : topicConsumerInfoList.get(0);
//				// 该主题该消费组下的所有队列，接下来就按broker分组
//				List<QueueStatInfo> queueStatInfoList = temp.getQueueStatInfoList();
//
//				// 接下来就按broker分组
//				Map<String/*this broker*/, Long/*this lag*/> mapToBrokerLag = new HashMap<>();
//
//				// 该主题该消费组下的所有队列
//				for (QueueStatInfo queueStatInfo : queueStatInfoList) {
//
//					String brokerAddr = brokerNameToAddrRMQ(clusterInfo, queueStatInfo.getBrokerName());
//					RMQAddrToName.put(brokerAddr, queueStatInfo.getBrokerName());
//					String key = topic + group + brokerAddr;
//
//					if (!mapToQueueOffset1.containsKey(key)) {
//						mapToQueueOffset1.put(key, new ArrayList<>());
//						mapToBrokerLag.put(brokerAddr, 0L);
//					}
//
//					// 后续第二次点击查询lag消息要用的
//					mapToQueueOffset1.get(key).add(new QueueOffset(queueStatInfo.getBrokerName(),
//						queueStatInfo.getBrokerAddr(),
//						queueStatInfo.getQueueId(),
//						queueStatInfo.getBrokerOffset(),
//						queueStatInfo.getConsumerOffset()));
//					mapToBrokerLag.put(brokerAddr, mapToBrokerLag.get(brokerAddr) + queueStatInfo.getBrokerOffset() - queueStatInfo.getConsumerOffset());
//				}
//
//				// 第三层遍历broker：统一一个TopicConsumerInfo的broker列表
//				for (String brokerAddr : mapToBrokerLag.keySet()) {
//					LagStatVO lagStatVO = new LagStatVO();
//					lagStatVO.setTopic(topic);
//					lagStatVO.setConsumerGroup(group);
//					lagStatVO.setBrokerAddr(brokerAddr);
//					lagStatVO.setBrokerName(RMQAddrToName.get(brokerAddr));
//					lagStatVO.setLag(mapToBrokerLag.get(brokerAddr));
//
//					lagStatVOList.add(lagStatVO);
//				}
//			}
//			return lagStatVOList;
//		} catch (Exception e) {
//			Throwables.throwIfUnchecked(e);
//			throw new RuntimeException(e);
//		}
//	}
//
//
//	/**
//	 * RMQ broker名转brokerAddr
//	 *
//	 * @param brokerName broker名
//	 * @return brokerAddr
//	 */
//	private String brokerNameToAddrRMQ(ClusterInfo clusterInfo, String brokerName) {
//		return clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr();
//	}
//
//
//	/**
//	 * 获取一个(topic, broker, consumerGroup)下lag消息 RMQ
//	 *
//	 * @param lagRecordQueryDTO 限定主题，限定group，限定broker，RMQ这里只有brokerName而不是地址
//	 * @return lag消息
//	 */
//	public PageResult getLagRecordListRMQ(LagRecordQueryDTO lagRecordQueryDTO) {
//		String topic = lagRecordQueryDTO.getTopic();
//		String group = lagRecordQueryDTO.getGroup();
//		String brokerAddr = lagRecordQueryDTO.getBrokerAddr();
//		log.error("获取主题: {} + 消费组: {} + broker: {} 的堆积消息...", topic, group, brokerAddr);
//
//		// 对应的brokerName
//		String brokerName = RMQAddrToName.get(brokerAddr);
//
//		// 该broker下的所有分区 RMQ的 之前存储的
//		List<QueueOffset> queueOffsetList = mapToQueueOffset1.get(topic + group + brokerAddr);
//
//		// TODO 排序
//		List<LagRecord> lagRecordList = new ArrayList<>();
//
//		// 一个辅助消费者，不影响其他消费组的消费
//		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(Constants.KeyStr.TEMP_GROUP_ID);
//		consumer.setNamesrvAddr(namesrvAddr);
//		try {
//			consumer.start();
//
//			// 遍历broker的每个分区
//			for (QueueOffset queueOffset : queueOffsetList) {
//
//				MessageQueue messageQueue = new MessageQueue(topic, queueOffset.getBrokerName(), queueOffset.getPartition());
//				long consumerOffset = queueOffset.getConsumerOffset();
//				long brokerOffset = queueOffset.getBrokerOffset();
//
//				for (long offset = consumerOffset + 1; offset <= brokerOffset; offset++) {
//					// 根据已有的offset数据，批量取未消费的消息 consumerOffset -> brokerOffset TODO 第四个参数long转int可能会有bug
//					PullResult pullResult = consumer.pull(messageQueue, "*", offset, 1);
//					if (pullResult != null) {
//						switch (pullResult.getPullStatus()) {
//							case FOUND:
//								// 读取结果
//								List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
//								MessageExt messageExt = msgFoundList.get(0);
//
//								LagRecord lagRecord = new LagRecord();
//								lagRecord.setOffset(messageExt.getQueueOffset());
//								lagRecord.setValue(new String(messageExt.getBody(), Charsets.UTF_8)); // 设置具体消息内容
//								lagRecordList.add(lagRecord);
//								lagRecord.setTimestamp(messageExt.getBornTimestamp());
//							case NO_MATCHED_MSG:
//							case NO_NEW_MSG:
//							case OFFSET_ILLEGAL:
//								log.error("Get Lag Msg Error! OFFSET_ILLEGAL");
//						}
//					}
//				}
//			}
//		} catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
//			log.error("获取{}的堆积消息失败！", lagRecordQueryDTO);
//			throw new RuntimeException(e);
//		} finally {
//			consumer.shutdown();
//		}
//		// TODO 排序
//		return new PageResult(lagRecordList.size(), lagRecordList);
//	}
//
//
//	/**
//	 * RMQ 进入页面首先展示所有主题所有消费组所有broker的消费情况(主题，broker，消费组地址，消费组名称，lag)
//	 *
//	 * @return (主题, 消费组, broker地址, broker名称, lag)
//	 */
//	public PageResult listAllConsumeStatRMQ() {
//		// 1. 列出所有topic
//		Set<String> topicList = new HashSet<>();
//		topicList = getTopicListRMQ(false, false);
//
//		List<LagStatVO> lagStatVOList = new ArrayList<>();
//
//		// 第一层遍历：遍历所有主题
//		for (String topic : topicList) {
//
//			lagStatVOList.addAll(getConsumeStatsListByTopicRMQ(topic));
//		}
//
//		// 返回总的消费情况结果
//		PageResult pageResult = new PageResult(lagStatVOList.size(), lagStatVOList);
//		return pageResult;
//	}
//}
